// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <algorithm>
#include <map>
#include <memory>
#include <string>
#include <vector>

#include "coin2/base/config/app_config.h"
#include "coin2/exchange/base/executor/topic_distributor.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/base/feed/processor.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/flow/writer.h"

namespace coin2::exchange::base::feed::live_writer {
namespace impl {

using coin2::exchange::feed::ArchiveConfig;

class FeedWriterProcessor : public executor::TopicRecordSubscriber {
 public:
  explicit FeedWriterProcessor(
      const MarketExchangeApi& mea,
      const ArchiveConfig& config,
      boost::asio::io_context* writer_ioc)
      : config_(config),
        writer_ioc_(writer_ioc) {
    CHECK(writer_ioc);
  }

  void OnTopicRecord(const executor::TopicRecordPacket& packet) override {
    // Bitmex has >32kb trade messages
    auto buf = std::make_shared<std::string>();
    buf->resize(orio::record::TopicRecord::struct_size_with_data(packet.payload.size()));
    auto* r = reinterpret_cast<orio::record::TopicRecord*>(buf->data());

    r->data_len = packet.payload.size();
    r->queue_seq = 0;
    r->topic_id = 0;
    r->topic_seq = packet.sequence;
    r->timestamp = packet.timestamp;
    const auto* data = reinterpret_cast<const std::byte*>(packet.payload.data());
    std::copy(data, data + r->data_len, r->data);

    if (writers_.find(packet.topic_name) == writers_.end()) {
      writer_strands_[packet.topic_name] =
          std::make_unique<boost::asio::io_context::strand>(*writer_ioc_);
      if (config_.auto_move()) {
        boost::asio::io_context::strand& strand = *writer_strands_[packet.topic_name];
        writers_[packet.topic_name] = std::make_unique<FlowWriter>(
          &strand,
          config_.flow_root_dir(),
          config_.machine(),
          packet.topic_name);
      } else {
        writers_[packet.topic_name] = std::make_unique<FlowWriter>(
            config_.flow_root_dir(),
            config_.machine(),
            packet.topic_name);
      }
    }

    FlowWriter* writer = writers_[packet.topic_name].get();
    boost::asio::io_context::strand& strand = *writer_strands_[packet.topic_name];
    boost::asio::post(strand, [buf, writer] {
      auto* r = reinterpret_cast<orio::record::TopicRecord*>(buf->data());
      writer->Write(*r);
    });
  }

 private:
  std::map<std::string, std::unique_ptr<FlowWriter>> writers_;
  const ArchiveConfig& config_;
  boost::asio::io_context* writer_ioc_;
  std::map<std::string, std::unique_ptr<boost::asio::io_context::strand>> writer_strands_;
};

}  // namespace impl

using impl::FeedWriterProcessor;
}  // namespace coin2::exchange::base::feed::live_writer
